package com.zhengb.flink.study.datastream.connector;

import com.alibaba.fastjson.JSONObject;
import com.zhengbo.flink.demo.base.constant.FlinkConstant;
import com.zhengbo.flink.demo.base.kafka.KafkaConfig;
import lombok.extern.slf4j.Slf4j;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer011;

import java.util.Collections;
import java.util.Properties;

/**
 * @author zhengbo
 * @date 2020/2/1
 */
@Slf4j
public class FlinkKafkaConsumerTest {

    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.enableCheckpointing(10000, CheckpointingMode.EXACTLY_ONCE);
        env.getCheckpointConfig().setCheckpointTimeout(100000);
        env.getCheckpointConfig().setMaxConcurrentCheckpoints(1);

        String topic = "neu_car";

        Properties properties = KafkaConfig.kafkaConsumerProperties(FlinkConstant.KAFKA_BROKER_LIST, "flink-test");

        FlinkKafkaConsumer011<String> kafkaConsumer011 = new FlinkKafkaConsumer011<String>(
                Collections.singletonList(topic), new SimpleStringSchema(), properties);
        kafkaConsumer011.setStartFromGroupOffsets();

        env.addSource(kafkaConsumer011).addSink(new RichSinkFunction<String>() {
            @Override
            public void invoke(String value, Context context) throws Exception {

                KafkaReceiveEntity entity = JSONObject.parseObject(value, KafkaReceiveEntity.class);

                log.info("sink kafka data:{}", JSONObject.toJSONString(entity));
            }
        });

        env.execute("FlinkKafkaConsumerTest");

    }
}
